/*
 *    Copyright (c) 2018-2025, lengleng All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 * Redistributions of source code must retain the above copyright notice,
 * this list of conditions and the following disclaimer.
 * Redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the following disclaimer in the
 * documentation and/or other materials provided with the distribution.
 * Neither the name of the pig4cloud.com developer nor the names of its
 * contributors may be used to endorse or promote products derived from
 * this software without specific prior written permission.
 * Author: lengleng (wangiegie@gmail.com)
 */


package com.pig4cloud.pigx.gateway.filter;

import cn.hutool.http.HttpUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.pig4cloud.pigx.common.core.constant.SecurityConstants;
import com.pig4cloud.pigx.common.core.util.R;
import com.pig4cloud.pigx.common.customize.api.exception.OpenApiException;
import com.pig4cloud.pigx.common.customize.api.util.OpenApiUtil;
import com.pig4cloud.pigx.common.customize.enums.CommonDictEnum;
import com.pig4cloud.pigx.common.customize.redis.constants.RedisKeyEnum;
import com.pig4cloud.pigx.user.api.feign.RippleRemoteGatewayService;
import com.pig4cloud.pigx.user.api.feign.RippleRemoteUserService;
import com.pig4cloud.pigx.user.enums.UserApiIpEnum;
import com.pig4cloud.pigx.user.enums.UserApiKeyEnum;
import com.pig4cloud.pigx.user.vo.UserApiIpCacheVO;
import com.pig4cloud.pigx.user.vo.UserApiKeyInfoVO;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.Strings;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;

/**
 * 开放Api全局过滤器
 *
 * @since 2022/11/8 16:57
 */
@Slf4j
@Component
public class RippleApiWsGlobalFilter implements GlobalFilter, Ordered {

    @Autowired
    @Qualifier("redisTemplate")
    private RedisTemplate redis;
    @Autowired
    private RippleRemoteGatewayService rippleRemoteGatewayService;
    @Autowired
    private RippleRemoteUserService rippleRemoteUserService;
    @Autowired
    private ObjectMapper objectMapper;

    /**
     * 网关统一处理API请求
     * <p> 重放攻击验证
     * <p> token鉴权
     *
     * @param exchange 路由
     * @param chain    过滤链
     * @return {@link Mono< Void>}
     * @since 2022/11/8 17:43
     */
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();

        // 清洗请求头
        clearHeaders(request);

        // 不是Api的ws请求直接放行
        String uri = request.getURI().getRawPath();
        if (!isApiWs(uri)) {
            // System.err.println("非开放Api的ws请求");
            return chain.filter(exchange);
        }

        try {
            // 处理请求
            handleRequest(request);
            // 去掉uri的/stream前缀
            String newUri = uri.replaceFirst("/stream", "");
            request = request.mutate().path(newUri).build();

            return chain.filter(exchange.mutate().request(request).build());
        } catch (Exception e) {
            // 处理异常
            return handleException(exchange, chain, e);
        }
    }

    /**
     * 处理请求
     *
     * @param request 请求包
     * @since 2022/11/8 17:51
     */
    public void handleRequest(ServerHttpRequest request) {
        // 获取请求头中的AccessKey
        String accessKey = getAccessKey(request);
        // 获取请求头中的客户端ip地址
        String clientIp = getClientIp(request);

        // 获取token
        String token = getToken(accessKey, clientIp);
        // System.err.println("token：" + token);
        // 设置headers
        setHeaders(request, token);

        // 根据AccessKey获取ApiKey信息
        UserApiKeyInfoVO apiKey = getApiKey(accessKey, token);

        // 校验ApiKey信息
        checkApiKey(clientIp, apiKey);
        // 校验访问频率
        handleRateLimit(apiKey, clientIp);
    }

    /**
     * 校验访问频率
     *
     * @param apiKey   apiKey信息
     * @param clientIp 客户端ip
     * @since 2022/11/13 17:03
     */
    private void handleRateLimit(UserApiKeyInfoVO apiKey, String clientIp) {
        String key = OpenApiUtil.getWsRateLimitRedisKey(apiKey.getAccessKey(), clientIp);
        Long count = redis.opsForValue().increment(key, 1);
        if (Objects.isNull(count)) {
            // 服务器内部错误
            throw new OpenApiException(10001, "Internal error; unable to process your request. Please try again.");
        }

        // 设置过期时间
        RedisKeyEnum wsRateLimit = RedisKeyEnum.ApiWsRateLimit;
        Long expire = redis.getExpire(key);
        if (Objects.isNull(expire) || expire == -1) {
            redis.expire(key, wsRateLimit.getExpire(), wsRateLimit.getExpireUnit());
        }

        // 校验连接次数，todo 改动可以动态设置的参数
        int maxConnectionCount = 100;
        if (count > maxConnectionCount) {
            // System.err.println("ws连接次数过多");
            throw new OpenApiException(10003, "Too many requests queued.");
            // todo 封号
        }
    }

    /**
     * 清洗headers
     *
     * @param request 请求包
     * @since 2022/11/10 16:47
     */
    private void clearHeaders(ServerHttpRequest request) {
        // 清洗传入的x-api-ws设置
        request = request.mutate()
                .headers(httpHeaders -> httpHeaders.remove(SecurityConstants.X_API_WS)).build();
    }

    /**
     * 设置headers
     *
     * @param request 请求包
     * @param token   token
     * @since 2022/11/10 16:47
     */
    private void setHeaders(ServerHttpRequest request, String token) {
        request = request.mutate()
                .headers(httpHeaders -> {
                    httpHeaders.add(SecurityConstants.AUTHORIZATION, token);
                    // 标记是开放API的ws请求
                    httpHeaders.add(SecurityConstants.X_API_WS, SecurityConstants.TRUE);
                }).build();
    }

    /**
     * 获取token
     *
     * @param accessKey 访问密钥
     * @param clientIp  clientIp
     * @since 2022/11/9 17:21
     */
    private String getToken(String accessKey, String clientIp) {

        // 尝试从缓存中获取token
        String key = OpenApiUtil.getTokenRedisKey(accessKey, clientIp);
        Object tokenObj = redis.opsForValue().get(key);
        if (!Objects.isNull(tokenObj)) {
            return tokenObj.toString();
        }

        // 如果没有，进行登陆
        HttpHeaders requestHeaders = new HttpHeaders();
        requestHeaders.set(HttpHeaders.AUTHORIZATION, HttpUtil.buildBasicAuth("api", "api", StandardCharsets.UTF_8));
        requestHeaders.set(SecurityConstants.X_API_KEY, accessKey);

        Map<String, Object> params = new HashMap<>(1);
        params.put("grant_type", "api");

        // webflux需要异步调用
        CompletableFuture<Map<String, Object>> future = CompletableFuture.supplyAsync(() -> rippleRemoteGatewayService.sendGateWay(requestHeaders, params, SecurityConstants.FROM_IN));
        Map<String, Object> res;
        try {
            res = future.get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        tokenObj = res.get(SecurityConstants.ACCESS_TOKEN);
        if (Objects.isNull(tokenObj)) {
            // 无效的AccessKey
            throw new OpenApiException(20004, "Invalid API-key, IP, or permissions for action.");
        }

        String token = SecurityConstants.BEARER + tokenObj.toString();

        // 缓存token
        RedisKeyEnum loginToken = RedisKeyEnum.ApiLoginToken;
        redis.opsForValue().set(key, token, loginToken.getExpire(), loginToken.getExpireUnit());
        return token;

    }

    /**
     * 检查ApiKey信息
     *
     * @param clientIp 客户端IP
     * @param apiKey   ApiKey信息
     * @since 2022/11/9 11:46
     */
    private void checkApiKey(String clientIp, UserApiKeyInfoVO apiKey) {

        // 为空 || 封禁状态
        if (Objects.isNull(apiKey) || !Objects.equals(UserApiKeyEnum.Status.NORMAL.getValue(), apiKey.getStatus())) {
            // 无效的AccessKey
            throw new OpenApiException(20004, "Invalid API-key, IP, or permissions for action.");
        }

        // 如果没有设置IP白名单，不再进行以下校验
        if (Objects.equals(CommonDictEnum.YesOrNo.NO.getValue(), apiKey.getHasIpLimit())) {
            return;
        }

        // 校验ip
        List<UserApiIpCacheVO> ipList = apiKey.getIpList();
        if (CollectionUtils.isEmpty(ipList)) {
            // 无效的IP
            throw new OpenApiException(20004, "Invalid API-key, IP, or permissions for action.");
        }

        // 校验IP白名单
        Optional<UserApiIpCacheVO> ipOpt = ipList.stream()
                .filter(i -> Objects.equals(i.getAddress(), clientIp))
                .findFirst();

        if (!ipOpt.isPresent()) {
            // 无效的IP
            throw new OpenApiException(10005, "No such IP has been white listed.");
        } else {
            UserApiIpCacheVO ip = ipOpt.get();
            if (!Objects.equals(UserApiIpEnum.Status.NORMAL.getValue(), ip.getStatus())) {
                // 无效的IP
                throw new OpenApiException(20004, "Invalid API-key, IP, or permissions for action.");
            }
        }

    }

    /**
     * 获取ApiKey信息
     *
     * @param accessKey 访问密钥
     * @param token     token
     * @return {@link UserApiKeyInfoVO}
     * @since 2022/11/9 11:46
     */
    private UserApiKeyInfoVO getApiKey(String accessKey, String token) {
        // webflux需要异步调用
        CompletableFuture<R<UserApiKeyInfoVO>> future = CompletableFuture.supplyAsync(() -> rippleRemoteUserService.getApiKeyInfo(accessKey, SecurityConstants.FROM_IN, token));
        R<UserApiKeyInfoVO> apiKeyR;
        try {
            apiKeyR = future.get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        if (!apiKeyR.getIsSuccess()) {
            // 服务器内部错误
            throw new OpenApiException(10001, "Internal error; unable to process your request. Please try again.");
        }

        return apiKeyR.getData();

    }

    /**
     * 从请求头中获取AccessKey
     *
     * @param request 请求包
     * @return {@link String}
     * @since 2022/11/8 17:59
     */
    private String getAccessKey(ServerHttpRequest request) {
        String accessKey = request.getHeaders().getFirst(SecurityConstants.X_API_KEY);
        if (Strings.isBlank(accessKey)) {
            // 无效的AccessKey
            throw new OpenApiException(20004, "Invalid API-key, IP, or permissions for action.");
        }

        return accessKey;
    }

    /**
     * 从请求头中获取用户Ip地址
     *
     * @param request 请求包
     * @return {@link String}
     * @since 2022/11/8 17:59
     */
    private String getClientIp(ServerHttpRequest request) {
        String xForwardedFor = request.getHeaders().getFirst(SecurityConstants.X_CLIENT_IP);
        return OpenApiUtil.getClientIp(xForwardedFor);
    }

    /**
     * 判断是否是Api的ws请求
     *
     * @return {@link boolean}
     * @since 2022/11/8 17:41
     */
    private boolean isApiWs(String uri) {
        return uri.startsWith("/stream");
    }

    /**
     * 标明优先级
     * <p>需要在PigxRequestGlobalFilter之前，否则路径中的/api会被清洗掉
     *
     * @return {@link int}
     * @since 2022/11/8 17:39
     */
    @Override
    public int getOrder() {
        return -1002;
    }

    /**
     * 处理异常
     *
     * @param exchange 路由
     * @param chain    过滤链
     * @param e        异常
     * @return {@link Mono< Void>}
     * @since 2022/11/10 16:37
     */
    private Mono<Void> handleException(ServerWebExchange exchange, GatewayFilterChain chain, Exception e) {
        ServerHttpResponse response = exchange.getResponse();

        response.getHeaders().setContentType(MediaType.APPLICATION_JSON);
        response.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);

        try {
            R<Object> r;
            e.printStackTrace();

            if (e instanceof OpenApiException) {
                OpenApiException apiException = (OpenApiException) e;
                r = R.failed(apiException.getCode(), apiException.getMessage());

            } else {
                r = R.failed(e.getMessage());
            }

            return response.writeWith(Mono.just(response.bufferFactory().wrap(objectMapper.writeValueAsBytes(r))));

        } catch (JsonProcessingException e1) {
            log.error("对象输出异常", e1);
            throw new OpenApiException(10001, "Internal error; unable to process your request. Please try again.");
        }
    }
}
